use anyhow::Error;

use indicatif::{ProgressBar, ProgressStyle};
use tokio::net::TcpStream;

use std::path::PathBuf;
use std::sync::{Arc};

use tokio::net::TcpListener;
use tokio::sync::mpsc::Sender;
use tokio::sync::oneshot;

use crate::args::Args;
use crate::network::session::Session;
use crate::network::session_manager::SessionManager;
use crate::network::session_role::SessionRole;
use crate::persistence::aof_file::{AofFile, SyncStrategy};
use crate::store::db::DatabaseMessage;
use crate::store::db_manager::DatabaseManager;
use crate::network::connection::Connection;
use crate::replication::ReplicationManager;
use crate::command::Command;
use crate::frame::Frame;

pub struct Server {
    args: Arc<Args>,
    aof_file: Option<AofFile>,
    aof_sender: Option<Sender<(usize, Frame)>>,
    session_manager: Arc<SessionManager>,
    db_manager: Arc<DatabaseManager>
}

impl Server {

    pub fn new(args: Arc<Args>) -> Self {
        let session_manager = Arc::new(SessionManager::new());
        let db_manager = Arc::new(DatabaseManager::new(args.clone()));
        let (aof_file, aof_sender) = if args.appendonly == "yes" {
            let file_path = PathBuf::from(&args.dir).join(&args.appendfilename);
            let sync_strategy = SyncStrategy::from_str(&args.appendfsync);
            let file = AofFile::new(file_path, sync_strategy);
            let sender = file.get_sender();
            (Some(file), Some(sender))
        } else {
            (None, None)
        };

        Server { 
            args, 
            aof_file, 
            aof_sender,
            session_manager,
            db_manager
        }
    }

    pub async fn start(&mut self) {

        if let Some(af) = &mut self.aof_file {
            if let Err(_) = Self::replay_aof_file(af, self.db_manager.clone()).await {
                log::info!("Failed to load AOF file");
            }
        }

        if self.args.is_slave() {
            let args = self.args.clone();
            let db_manager = self.db_manager.clone();
            std::thread::spawn(move || {
                let rt = tokio::runtime::Runtime::new().unwrap();
                rt.block_on(async {
                    let mut rm = ReplicationManager::new(args,  db_manager);
                    if let Err(e) = rm.connect().await {
                        log::error!("Failed to connect to master: {}", e);
                    }
                });
            });
        } 

        match TcpListener::bind(format!("{}:{}", self.args.bind, self.args.port)).await {
            Ok(listener) => {
                log::info!("Server initialized");
                log::info!("Ready to accept connections");
                loop {
                    match listener.accept().await {
                        Ok((stream, _address)) => {
                            
                            // 检查 maxclients 限制
                            if self.session_manager.is_over_max_clients(self.args.maxclients) {
                                let connection = crate::network::connection::Connection::new(stream);
                                let error_frame = crate::frame::Frame::Error("ERR max number of clients reached".to_string());
                                tokio::spawn(async move {
                                    connection.write_bytes(error_frame.as_bytes()).await;
                                });
                                continue;
                            }
                            
                            let aof_sender = self.aof_sender.clone(); 
                            let session_manager_clone = self.session_manager.clone();
                            let db_manager_clone = self.db_manager.clone();
                            let mut handler = Handler::new(db_manager_clone, session_manager_clone, stream, self.args.clone(), aof_sender);
                            tokio::spawn(async move {
                                handler.handle().await;
                            });
                        }
                        Err(e) => {
                            log::error!("Failed to accept connection: {}", e);
                        }
                    }
                }
            }
            Err(_e) => {
                log::error!("Failed to bind to address {}:{}", self.args.bind, self.args.port);
                std::process::exit(1);
            }
        }
    }

    async fn replay_aof_file(aof_file: &mut AofFile, db_manager: Arc<DatabaseManager>) -> Result<(), Error>  {
        let frames = aof_file.read_all_frames().await.unwrap();
        let pb = ProgressBar::new(frames.len() as u64);
        pb.set_style(ProgressStyle::default_bar()
            .template("{spinner:.green} [{bar:40.green/gray}] {pos}/{len} ({percent}%) {msg}")
            .tick_strings(&["⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷"])
            .progress_chars("█▓▒░")
        );
        pb.set_message("Status: In progress");
        let mut current_db_index = 0;
        for frame in frames {
            let command = match Command::parse_from_frame(frame) {
                Ok(cmd) => cmd,
                Err(e) => {
                    log::warn!("Skipping invalid frame in AOF: {}", e);
                    pb.inc(1);
                    continue; 
                }
            };
            match command {
                Command::Select(select) => {
                    current_db_index = select.get_db_index();
                },
                _ => {
                    let db_sender = db_manager.get_sender(current_db_index);
                    let (sender, receiver) = oneshot::channel();
                    let message = DatabaseMessage::Command { sender, command };
                    if let Err(e) = db_sender.send(message).await {
                        log::warn!("Failed to send command to database during AOF replay: {}", e);
                    } else {
                        let _ = receiver.await;
                    }
                }
            }
            pb.inc(1);
        }
        pb.set_message("Status: Completed");
        pb.finish();
        println!();
        Ok(())
    }
}

pub struct Handler {
    session: Session,
    aof_sender: Option<Sender<(usize, Frame)>>,
    session_manager: Arc<SessionManager>,
    db_manager: Arc<DatabaseManager>,
    args: Arc<Args>
}

impl Handler {
    pub fn get_session(&self) -> &Session {
        &self.session
    }
    
    pub fn get_db_manager(&self) -> &Arc<DatabaseManager> {
        &self.db_manager
    }
    
    pub fn get_args(&self) -> &Arc<Args> {
        &self.args
    }
}

impl Handler {

    pub fn new(db_manager: Arc<DatabaseManager>, session_manager: Arc<SessionManager>, stream: TcpStream, args: Arc<Args>, aof_sender: Option<Sender<(usize,Frame)>>) -> Self {
        let args_ref = args.as_ref();
        let certification = args_ref.requirepass.is_none();
        let sender = db_manager.as_ref().get_sender(0);
        let connection = Connection::new(stream);
        let session = Session::new(certification, sender, connection);
        session_manager.create_session(session.clone());

        Handler {
            session,
            aof_sender,
            session_manager,
            db_manager,
            args,
        }
    }

    /**
     * 客户端登录认证
     * 
     * 如果 "密码" 不匹配，响应 ERR invalid password 错误
     * 
     * @param input_requirepass 输入密码【只读】
     */
    pub fn login(&mut self, input_requirepass: &String) -> Result<(), Error> {
        if let Some(ref requirepass) = self.args.requirepass {
            if requirepass == input_requirepass {
                self.session.set_certification(true);
                return Ok(())
            } 
            return Err(Error::msg("ERR invalid password"));
        } else {
            Ok(())
        }
    }

    /**
     * 切换当前数据库索引
     * 
     * 如果索引超出，响应 ERR DB index is out of range 错误
     * 
     * @param idx 目标数据库索引
     */
    pub fn change_sender(&mut self, idx: usize) -> Result<(), Error> {
        if self.args.databases - 1 < idx {
            return Err(Error::msg("ERR DB index is out of range"));
        }
        self.session.set_current_db(idx);
        self.session.set_sender(self.db_manager.get_sender(idx));
        Ok(())
    }

    /**
     * 设置 SessionRole 并同步到 SessionManager
     * 
     * @param role 会话角色
     */
    pub fn set_session_role(&mut self, role: SessionRole) {
        self.session.set_role(role);
        self.session_manager.create_session(self.session.clone());
    }

    /// Handling client connections
    pub async fn handle(&mut self) {
        loop {

            log::debug!("Waiting for bytes");
            let bytes = match self.session.connection.read_bytes().await {
                Ok(bytes) => bytes,
                Err(_e) => {
                    self.session_manager.remove_session(self.session.get_id());
                    return;
                }
            };
            
            // 解析可能的多个粘连命令帧
            let frames = match Frame::parse_multiple_frames(bytes.as_slice()) {
                Ok(frames) => frames,
                Err(e) => {
                    log::error!("Failed to parse multiple frames: {:?}", e);
                    let frame = Frame::Error(format!("Failed to parse frames: {:?}", e));
                    self.session.connection.write_bytes(frame.as_bytes()).await;
                    continue;
                }
            };
            
            log::debug!("Received bytes: {:?}", String::from_utf8_lossy(bytes.as_slice()));
            
            for frame in frames {
                log::debug!("Received frame: {}", frame.to_string());
                let frame_copy = frame.clone();
                if self.session.is_in_transaction() {
                    let command_name = frame.get_arg(0).unwrap_or_default().to_uppercase();
                    if command_name != "EXEC" && command_name != "DISCARD" {
                        self.session.add_transaction_frame(frame_copy);
                        self.session.connection.write_bytes(Frame::SimpleString("QUEUED".to_string()).as_bytes()).await;
                        continue;
                    }
                }
                
                let command = match Command::parse_from_frame(frame) {
                    Ok(cmd) => cmd,
                    Err(e) => {
                        let frame = Frame::Error(e.to_string());
                        self.session.connection.write_bytes(frame.as_bytes()).await;
                        continue;
                    }
                };
                
                match command {
                    Command::Auth(_) => {},
                    _ => { 
                        if self.args.requirepass.is_some() {
                            if self.session.get_certification() == false {
                                let frame = Frame::Error("NOAUTH Authentication required.".to_string());
                                self.session.connection.write_bytes(frame.as_bytes()).await;
                                continue;
                            }
                        } 
                    },
                };

                let is_psync_command = matches!(command, Command::Psync(_));
                let should_propagate = command.propagate_aof_if_needed();
                let result = self.apply_command(command).await;

                match result {
                    Ok(frame) => {
                        if should_propagate {
                            if let Some(ref aof_sender) = self.aof_sender {
                                let _ = aof_sender.send((self.session.get_current_db(), frame_copy.clone())).await;
                            }
                            self.propagate_to_slaves(frame_copy.clone()).await;
                        }
                        self.session.connection.write_bytes(frame.as_bytes()).await;
                        if is_psync_command {
                            return;
                        }
                    }
                    Err(e) => {
                        println!("Failed to receive; err = {:?}", e);
                    }
                }
            }
        }
    }
    
    /// 执行服务器命令
    async fn apply_command(&mut self, command: Command) -> Result<Frame, Error> {
        match command {
            Command::Auth(auth) => auth.apply(self),
            Command::Client(client) => client.apply(),
            Command::Replconf(replconf) => replconf.apply(self),
            Command::Save(save) => save.apply(self.db_manager.clone(), self.args.clone()).await,
            Command::Bgsave(bgsave) => bgsave.apply(self.db_manager.clone(), self.args.clone()).await,
            Command::Psync(psync) => psync.apply(self.db_manager.clone(), self.args.clone()).await,
            Command::Flushall(flushall) => flushall.apply(self.db_manager.clone()).await,
            Command::Move(r#move) => r#move.apply(self).await,
            Command::Exec(_) => Box::pin(self.execute_transaction()).await,
            Command::Multi(multi) => multi.apply(self),
            Command::Discard(discard) => discard.apply(self),
            Command::Select(select) => select.apply(self),
            Command::Unknown(unknown) => unknown.apply(),
            Command::Ping(ping) => ping.apply(),
            Command::Echo(echo) => echo.apply(),
            _ => self.apply_db_command(command).await,
        }
    }

    /// 执行事务中的所有命令
    async fn execute_transaction(&mut self) -> Result<Frame, Error> {

        if !self.session.is_in_transaction() {
            return Ok(Frame::Error("ERR EXEC without MULTI".to_string()));
        }

        let transaction_frames = self.session.get_transaction_frames().clone();
        let mut results = Vec::new();
        for frame in transaction_frames {
            let command = match Command::parse_from_frame(frame) {
                Ok(cmd) => cmd,
                Err(e) => {
                    results.push(Frame::Error(e.to_string()));
                    continue;
                }
            };
            
            // 应用命令并收集结果
            // 注意：这里我们不处理EXEC、MULTI、DISCARD命令，避免递归
            match command {
                Command::Exec(_) | Command::Multi(_) | Command::Discard(_) => {
                    results.push(Frame::Error("ERR nested transaction commands not allowed".to_string()));
                },
                _ => {
                    // 为了避免递归（实际不会有, 解决 Rust 编译问题）
                    let result = match command {
                        Command::Auth(auth) => auth.apply(self),
                        Command::Client(client) => client.apply(),
                        Command::Replconf(replconf) => replconf.apply(self),
                        Command::Save(save) => save.apply(self.db_manager.clone(), self.args.clone()).await,
                        Command::Bgsave(bgsave) => bgsave.apply(self.db_manager.clone(), self.args.clone()).await,
                        Command::Psync(psync) => psync.apply(self.db_manager.clone(), self.args.clone()).await,
                        Command::Flushall(flushall) => flushall.apply(self.db_manager.clone()).await,
                        Command::Move(r#move) => r#move.apply(self).await,
                        Command::Select(select) => select.apply(self),
                        Command::Unknown(unknown) => unknown.apply(),
                        Command::Ping(ping) => ping.apply(),
                        Command::Echo(echo) => echo.apply(),
                        _ => self.apply_db_command(command).await,
                    };
                    match result {
                        Ok(frame) => results.push(frame),
                        Err(e) => results.push(Frame::Error(e.to_string())),
                    }
                }
            }
        }
        self.session.clear_transaction();
        Ok(Frame::Array(results))
    }

    /// 执行数据库命令
    async fn apply_db_command(&self, command: Command) -> Result<Frame, Error> {
        let (sender, receiver) = oneshot::channel();
        let message = DatabaseMessage::Command { sender, command };
        let db_sender = self.session.get_sender();
        if let Err(e) = db_sender.send(message).await {
            return Ok(Frame::Error(format!("Channel closed: {:?}", e)));
        }
        let result = match receiver.await {
            Ok(f) => f,
            Err(e) => Frame::Error(format!("{:?}", e))
        };
        Ok(result)
    }

    /// 传播主节点命令
    async fn propagate_to_slaves(&self, frame: Frame) {
        let slave_sessions = self.session_manager.get_slave_sessions();
        let current_db = self.session.get_current_db();
        if slave_sessions.is_empty() {
            return;
        }
        
        for slave_session in slave_sessions {
            let frame_clone = frame.clone();
            let db = current_db;
            
            slave_session.connection.write_bytes(Frame::Array(vec![Frame::BulkString("SELECT".to_string()),Frame::BulkString(db.to_string())]).as_bytes()).await;
            slave_session.connection.write_bytes(frame_clone.as_bytes()).await;
        }
    }

    // 事务相关方法
    pub fn start_transaction(&mut self) {
        self.session.start_transaction();
    }

    pub fn is_in_transaction(&self) -> bool {
        self.session.is_in_transaction()
    }

    pub fn add_transaction_frame(&mut self, frame: Frame) {
        self.session.add_transaction_frame(frame);
    }

    pub fn get_transaction_frames(&self) -> Vec<Frame> {
        self.session.get_transaction_frames().clone()
    }

    pub fn clear_transaction(&mut self) {
        self.session.clear_transaction();
    }


}